package kafka.demo;

import kafka.KafkaProperties;
import kafka.api.FetchRequest;
import kafka.api.FetchRequestBuilder;
import kafka.javaapi.FetchResponse;
import kafka.javaapi.consumer.SimpleConsumer;
import kafka.javaapi.message.ByteBufferMessageSet;
import kafka.message.MessageAndOffset;

import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;

/**
 * @author: pizhihui
 * @datae: 2017-06-12
 */
public class SimpleConsumerDemo {

    private static void printMessages(ByteBufferMessageSet messageSet) throws UnsupportedEncodingException {
        for (MessageAndOffset messageAndOffset : messageSet) {
            ByteBuffer payload = messageAndOffset.message().payload();
            byte[] bytes = new byte[payload.limit()];
            payload.get(bytes);
            System.out.println(new String(bytes, "UTF-8"));
        }
    }

    public static void main(String[] args) throws UnsupportedEncodingException {

        SimpleConsumer simpleConsumer = new SimpleConsumer(KafkaProperties.KAFKA_SERVER_URL,
                KafkaProperties.KAFKA_SERVER_PORT,
                KafkaProperties.CONNECTION_TIMEOUT,
                KafkaProperties.KAFKA_PRODUCER_BUFFER_SIZE,
                KafkaProperties.CLIENT_ID);

        FetchRequest request = new FetchRequestBuilder()
                .clientId(KafkaProperties.CLIENT_ID)
                .addFetch(KafkaProperties.TOPIC_NAME, 0, 0L, 100)
                .build();
        FetchResponse response = simpleConsumer.fetch(request);

        ByteBufferMessageSet messageSet = response.messageSet(KafkaProperties.TOPIC_NAME, 0);
        printMessages(messageSet);


    }

}
